//
// Created by martin on 2/16/22.
//

#include "TcpConnection.h"
#include "muduo/base/Logging.h"
#include "muduo/base/WeakCallback.h"
#include "muduo/net/Socket.h"
#include "muduo/net/Channel.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/Buffer.h"
#include "muduo/net/SocketsOps.h"

#include <functional>

using namespace muduo;
using namespace muduo::net;

void muduo::net::defaultConnectionCallback(const TcpConnectionPtr& conn)
{
	LOG_TRACE << conn->localAddress().toIpPort() << " -> "
		<< conn->peerAddress().toIpPort() << " is "
		<< (conn->connected() ? "UP" : "DOWN");
	// do not call conn->forceClose(), because some users want to register message callback only.
}

void muduo::net::defaultMessageCallback(const TcpConnectionPtr& conn,
	Buffer* buf,
	Timestamp receiveTime)
{
	buf->retrieveAll();
}

TcpConnection::TcpConnection(EventLoop* loop,
	const std::string& nameArg,
	int sockfd,
	const InetAddress& localAddr,
	const InetAddress& peerAddr)
	: loop_(CHECK_NOTNULL(loop)),
	name_(nameArg),
	state_(kConnecting),
	reading_(true),
	socket_(new Socket(sockfd)),         // 绑定sockfd, 用于获取底层tcp协议栈信息
	channel_(new Channel(loop, sockfd)), // 绑定sockfd, 用于监听事件, 回调处理事件
	localAddr_(localAddr), // 本地地址, 包含ip, port, 协议类型等地址信息
	peerAddr_(peerAddr),   // 对端地址, 包含ip, port, 协议类型等地址信息
	highWaterMark_(64 * 1024 * 1024) // 高水位, 当outputBuffer中缓存数据超过高水位时, 就会触发高水位回调, 通知用户
{
	// 设置Channel事件回调
	channel_->setReadCallback(std::bind(&TcpConnection::handleRead, this, _1));
	channel_->setWriteCallback(std::bind(&TcpConnection::handleWrite, this));
	channel_->setCloseCallback(std::bind(&TcpConnection::handleClose, this));
	channel_->setErrorCallback(std::bind(&TcpConnection::handleError, this));
	LOG_DEBUG << "TcpConnection::ctor[" << name_ << "] at " << this
		<< " fd=" << sockfd;
	socket_->setKeepAlive(true);
}

TcpConnection::~TcpConnection()
{
	LOG_INFO << "TcpConnection::dtor[" << name_ << "] at " << this
		<< " fd=" << channel_->fd()
		<< " state=" << stateToString();
	assert(state_ == kDisconnected);
}

bool TcpConnection::getTcpInfo(struct tcp_info* tcpi) const
{
	return socket_->getTcpInfo(tcpi);
}

string TcpConnection::getTcpInfoString() const
{
	char buf[1024];
	buf[0] = '\0';
	socket_->getTcpInfoString(buf, sizeof(buf));
	return buf;
}

/**
 * 发送消息给对端, 允许在其他线程调用
 * @param message 要发送的消息. StringPiece兼容C/C++风格字符串, 二进制缓存, 提供统一字符串接口
 */
void TcpConnection::send(const StringPiece& message)
{
	if (state_ == kConnected)
	{
		if (loop_->isInLoopThread())
		{ // 当前线程是所属loop线程
			sendInLoop(message);
		}
		else
		{ // 当前线程并非所属loop线程
			void (TcpConnection:: * fp)(const StringPiece & message);
			fp = &TcpConnection::sendInLoop;
			loop_->runInLoop(
				std::bind(fp,
					this, // FIXME
					message.as_string()));
		}
	}
}

#if 0
void TcpConnection::send(string&& mesage)
{
	send(StringPiece(mesage));
}
#endif

void TcpConnection::send(const void* message, int len)
{
	send(StringPiece(static_cast<const char*>(message), len));
}

// FIXME efficiency!!!
void TcpConnection::send(Buffer* buf)
{
	if (state_ == kConnected)
	{
		if (loop_->isInLoopThread())
		{
			// send all readable bytes
			sendInLoop(buf->peek(), buf->readableBytes());
			buf->retrieveAll();
		}
		else
		{
			void (TcpConnection:: * fp)(const StringPiece & message) = &TcpConnection::sendInLoop;
			loop_->runInLoop(std::bind(fp, shared_from_this(), buf->retrieveAllAsString()));
		}
	}
}

// FIXME: add by martin
void TcpConnection::send(Buffer&& message)
{
	Buffer tmp(message);
	send(&tmp);
}

void TcpConnection::sendInLoop(const StringPiece& message)
{
	sendInLoop(message.data(), static_cast<size_t>(message.size()));
}

void TcpConnection::sendInLoop(string&& message)
{
	sendInLoop(message.data(), static_cast<size_t>(message.size()));
}

/**
 * 在所属loop线程中, 发送data[len]
 *
 * @param data 要发送的缓冲区首地址
 *
 * @param len　要发送的缓冲区大小(bytes)
 *
 * @details 发生write错误, 如果发送缓冲区未满,　对端已发FIN/RST分节 表明tcp连接发生致命错误(faultError为true).
 * 函数并不会缓存缓冲区data数据, 如果有数据超过write一次, 需要调用者维护缓冲区生命周期
 */
void TcpConnection::sendInLoop(const char* data, size_t len)
{
	loop_->assertInLoopThread();
	ssize_t nwrote = 0;
	size_t remaining = len;
	bool faultError = false;
	if (state_ == kDisconnected) // 如果已经断开连接(kDisconnected), 就无需发送, 打印log(LOG_WARN)
	{
		LOG_WARN << "disconnected, give up writing";
		return;
	}

	// write一次, 往对端发送数据, 后面再看是否发生错误, 是否需要高水位回调
	// if no thing output queue, try writing directly
	if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
	{ // 如果通道没有使能监听写事件, 并且outputBuffer　没有待发送数据, 就直接通过socket写
		nwrote = sockets::write(channel_->fd(), data, len);
		if (nwrote >= 0)
		{
			remaining = len - static_cast<size_t>(nwrote);
			// 写完了data[len]到到内核缓冲区, 就回调writeCompleteCallback_
			if (remaining == 0 && writeCompleteCallback_)
			{
				loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
			}
		}
		else // nwrote < 0, error
		{
			nwrote = 0;
			if (errno != EWOULDBLOCK) // EWOULDBLOCK: 发送缓冲区已满, 且fd已设为nonblocking
			{ // O_NONBLOCK fd, write block but return EWOULDBLOCK error
				LOG_SYSERR << "TcpConnection::sendInLoop";
				if (errno == EPIPE || errno == ECONNRESET) // FIXME: any others?
				{ // EPIPE: reading end is closed; ECONNRESET: connection resetRequest by peer
					faultError = true;
				}
			}
		}
	}

	// 处理剩余待发送数据
	// 没有错误, 并且还有没有写完的数据, 说明内核发送缓冲区满, 要将未写完的数据添加到output buffer中
	assert(remaining <= len);
	if (!faultError && remaining > 0) // 没有故障, 并且还有待发送数据, 可能是发送太快, 对方来不及接收
	{ // no error and data remaining to be written
		size_t oldLen = outputBuffer_.readableBytes(); // Buffer中待发送数据(readable空间数据)

		// 如果readable空间数据 + 未写完数据, 超过highWaterMark_(高水位), 回调highWaterMarkCallback_
		if (oldLen + remaining >= highWaterMark_
			&& oldLen < highWaterMark_
			&& highWaterMarkCallback_)
		{
			loop_->queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
		}
		// append data to be written to the output buffer
		outputBuffer_.append(static_cast<const char*>(data) + nwrote, remaining);
		// enable write event for channel_
		if (!channel_->isWriting())
		{ // 如果没有在监听通道写事件, 就使能通道写事件
			channel_->enableWriting();
		}
	}
}

/**
 * 关闭连接写方向, 只有已连接状态才有效
 * @note 调用前, 建议先调用channel_->disableWriting()/disableAll(), 确保没有监听写事件
 */
void TcpConnection::shutdown()
{
	// FIXME: use compare and swap
	if (state_ == kConnected)
	{
		setState(kDisconnecting);
		// FIXME: shared_from_this()?
		loop_->runInLoop(std::bind(&TcpConnection::shutDownInLoop, shared_from_this()));
	}
}

void TcpConnection::shutDownInLoop()
{
	loop_->assertInLoopThread();
	if (!channel_->isWriting())
	{
		// we are not writing
		socket_->shutdownWrite();
	}
}

/**
 * 强制关闭连接, 只对连接为kConnected或kDisconnecting状态才有效
 * @details 为防止意外, 动作应该放到loop末尾去做
 */
void TcpConnection::forceClose()
{
	// FIXME: use compare and swap
	if (state_ == kConnected || state_ == kDisconnecting)
	{
		setState(kDisconnecting);
		loop_->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
	}
}

/**
 * 强制延时一段时间后关闭连接, 只对连接为kConnected或kDisconnecting状态才有效
 * @param seconds 要延时关闭的秒数
 */
void TcpConnection::forceCloseWithDelay(double seconds)
{
	if (state_ == kConnected || state_ == kDisconnecting)
	{
		setState(kDisconnecting);
		loop_->runAfter(seconds,
			makeWeakCallback(shared_from_this(),
				&TcpConnection::forceClose)); // not forceCloseInLoop to avoid race condition
	}
}

/**
 * 在所属loop循环中强制关闭连接, 只对连接为kConnected或kDisconnecting状态才有效
 */
void TcpConnection::forceCloseInLoop()
{
	loop_->assertInLoopThread();
	if (state_ == kConnected || state_ == kDisconnecting)
	{
		// as if we received 0 byte in handleRead()
		handleClose();
	}
}

void TcpConnection::shutDownAndForceCloseInLoop(double seconds)
{
	shutdown();
	forceCloseWithDelay(seconds);
}

const char* TcpConnection::stateToString() const
{
	switch (state_)
	{
	case kDisconnected:
		return "kDisconnected";
	case kConnecting:
		return "kConnecting";
	case kConnected:
		return "kConnected";
	case kDisconnecting:
		return "kDisconnecting";
	default:
		return "unknown state";
	}
}

void TcpConnection::setTcpNoDelay(bool on)
{
	socket_->setTcpNoDelay(on);
}

void TcpConnection::startRead()
{
	loop_->runInLoop(std::bind(&TcpConnection::startReadInLoop, this));
}

void TcpConnection::startReadInLoop()
{
	loop_->assertInLoopThread();
	if (!reading_ || !channel_->isReading())
	{
		channel_->enableReading();
		reading_ = true;
	}
}

void TcpConnection::stopRead()
{
	loop_->runInLoop(std::bind(&TcpConnection::stopReadInLoop, this));
}

void TcpConnection::stopReadInLoop()
{
	loop_->assertInLoopThread();
	if (reading_ || channel_->isReading())
	{
		channel_->disableReading();
		reading_ = false;
	}
}

/**
 * TcpServer/TcpClient 调用newConnection创建TcpConnection对象后, 用来做一些连接以建立的事后工作
 */
void TcpConnection::connectEstablished()
{
	loop_->assertInLoopThread();
	assert(state_ == kConnecting);
	setState(kConnected);
	channel_->tie(shared_from_this()); // tie this TcpConnection object to channel_
	channel_->enableReading(); // 使能监听读事件, 连接上有读事件发生时, 如收到对端数据, 会触发channel_::handleRead

	// FIXME: add by Martin
	assert(connectionCallback_ != nullptr); // connectionCallback_通常不允许为空, TcpServer/TcpClient可以设置为缺省值defaultConnectionCallback

	connectionCallback_(shared_from_this()); // 回调connectionCallback_
}

/**
 * TcpServer/TcpClient移除通道事件时, 调用该函数销毁tcp连接
 * @note 只有处于已连接状态(kConnected)的tcp连接, 才需要先更新状态, 关闭通道事件监听
 */
void TcpConnection::connectDestroyed()
{
	loop_->assertInLoopThread();
	if (state_ == kConnected) // 只有kConnected的连接, 才有必要采取断开连接动作
	{
		setState(kDisconnected);
		channel_->disableAll(); // 关闭通道事件监听

		connectionCallback_(shared_from_this()); // 调用连接回调
	}
	channel_->remove(); // 从EventLoop和Poller中移除监听通道事件
}

/**
 * 从输入缓存inputBuffer_读取数据, 交给回调messageCallback_处理
 * @param receiveTime 接收到读事件的时间点
 * @details 通常是TcpServer/TcpClient运行回调messageCallback_, 将处理机会传递给用户
 */
void TcpConnection::handleRead(Timestamp receiveTime)
{
	loop_->assertInLoopThread();
	int savedErrno = 0;
	ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno); // 从指定fd读取数据到内部缓冲
	if (n > 0)
	{
		messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
	}
	else if (n == 0)
	{
		handleClose();
	}
	else
	{
		errno = savedErrno;
		LOG_SYSERR << "TcpConnection::handleRead";
		handleError();
	}
}

/**
 * 内核发送缓冲区有空间了, 回调该函数.
 * @details 在send()之后触发. send()通常由用户主动调用, handleWrite由epoll_wait/poll监听,
 * 由EventLoop对应IO线程回调.
 */
void TcpConnection::handleWrite()
{
	loop_->assertInLoopThread();
	if (channel_->isWriting())
	{ // 只有通道在监听写事件时, 处理write事件才有意义
		// 继续往内核缓冲区写readable空间的待发送数据
		ssize_t n = sockets::write(channel_->fd(),
			outputBuffer_.peek(),
			outputBuffer_.readableBytes());
		if (n > 0)
		{
			outputBuffer_.retrieve(static_cast<size_t>(n));
			if (outputBuffer_.readableBytes() == 0)
			{ // 发送缓冲区已清空
				channel_->disableWriting();
				if (writeCompleteCallback_)
				{ // 应用层发送缓冲区已被清空, 就回调writeCompleteCallback_
					loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
				}
				if (state_ == kDisconnecting)
				{ // 发送缓冲区已清空, 并且连接状态是kDisconnecting, 要关闭连接
					shutDownInLoop(); // 关闭连接 写方向
				}
			}
		}
		else
		{
			LOG_SYSERR << "TcpConnection::handleWrite";
		}
	}
}

/**
 * 处理Tcp连接关闭
 * @details 更新状态为kDisconnected, 清除所有事件通道监听.
 * @note 必须在所属loop线程中运行.
 */
void TcpConnection::handleClose()
{
	loop_->assertInLoopThread(); // 确保在所属loop线程中运行
	LOG_TRACE << "fd = " << channel_->fd() << " state = " << stateToString();
	assert(state_ == kConnected || state_ == kDisconnecting);
	// we don't close fd, leave it to dtor, so we can find leaks easily.
	setState(kDisconnected); // 更新Tcp连接状态
	channel_->disableAll();     // 停止监听所有通道事件(读写事件)

	TcpConnectionPtr guardThis(shared_from_this());
	connectionCallback_(guardThis); // 连接回调
	// must be the last line
	closeCallback_(guardThis);      //　关闭连接回调, 给用户在处理close连接事件时, 做一些处理的机会
}

/**
 * 处理tcp连接错误, 打印错误log
 * @details 从tcp协议栈获取错误信息
 */
void TcpConnection::handleError()
{
	int err = sockets::getSocketError(channel_->fd());
	LOG_ERROR << "TcpConnection::handleError [" << name_
		<< "] - SO_ERROR = " << err << " " << strerror_tl(err);
}
